package SQL_L

import org.apache.spark.sql.SparkSession
import org.junit.Test

class JoinProcessor {
  val spark = SparkSession.builder()
    .master("local[*]")
    .appName("JoinProcessor")
    .getOrCreate()

  import spark.implicits._
  import org.apache.spark.sql.functions._

  val person = Seq((0, "Lucy", 0), (1, "Lily", 0), (2, "Tim", 2), (3, "Danial", 0))
    .toDF("id", "name", "cityId")
  person.createOrReplaceTempView("person")

  val cities = Seq((0, "Beijing"), (1, "Shanghai"), (2, "Guangzhou"))
    .toDF("id", "name")
  cities.createOrReplaceTempView("cities")

  @Test
  def introJoin() = {
    val df = person.join(cities, person.col("cityId") === cities.col("id"))
      .select(person.col("id"),
        person.col("name"),
        cities.col("name") as "city"
      )
    df.createOrReplaceTempView("user_city")
    spark.sql("select id,name,city from user_city where city = 'Beijing'").show()
  }

  /**
   * 交叉连接
   */
  @Test
  def cross() = {
    //使用cross join必须要有一个where，否则数据会成指数上升
    person.crossJoin(cities)
      .where(person.col("cityId") === cities.col("id"))
      .show()
    spark.sql("select u.id,u.name,c.name from person u cross join cities c where u.cityId = c.id").show()
  }

  /**
   * 内连接
   */
  @Test
  def inner() = {
    person.join(cities,
      person.col("cityId") === cities.col("id")
      , joinType = "inner")
      .show()
    spark.sql("select u.id,u.name,c.name from person u inner join cities c where u.cityId = c.id").show()
  }

  /**
   * 全外连接
   */
  @Test
  def fullouter() = {
    person.join(cities,
      person.col("cityId") === cities.col("id")
      , joinType = "full")
      .show()
    spark.sql("select u.id,u.name,c.name from person u full join cities c where u.cityId = c.id").show()
  }

  /**
   * 左外连接和右外连接
   */
  @Test
  def leftAndRight() = {
    //左
    person.join(cities,
      person.col("cityId") === cities.col("id")
      , joinType = "left")
      .show()
    spark.sql("select u.id,u.name,c.name from person u left join cities c where u.cityId = c.id").show()

    //右
    person.join(cities,
      person.col("cityId") === cities.col("id")
      , joinType = "right")
      .show()
    spark.sql("select u.id,u.name,c.name from person u right join cities c where u.cityId = c.id").show()
  }
  @Test
  def leftAntiAndSemi() = {
    person.join(right = cities,
      joinExprs = person("cityId") === cities("id"),
      joinType = "left_anti")
      .show()
    person.join(right = cities,
      joinExprs = person("cityId") === cities("id"),
      joinType = "left_semi")
      .show()
  }
}
